package worker

import (
	"encoding/json"
	"errors"
	"executeserver/src/model"
	"fmt"
	"log"
	"regexp"
	"strings"
	"time"
)

type TaskInfo struct {
	Id                 int
	Name               string
	SrcConnectionStr   string
	DestConnectionStr  string
	QueryBaseSql       string
	ScriptId           string
	LeaderIP           string
	TotalExecutionTime int
	TotalSize          int
	PageSize           int
	StartPage          int
	EndPage            int
	DataDestination    string
	DataVersion        int
	ServantName        string
	DataProcessingType int
	DataSql            string
	CaseSql            []string
	TaskEndTime        time.Time
}
type WorkInfo struct {
	IsLeader    bool
	TaskInfo    *TaskInfo
	Status      int
	UndonePages []int
	Page        int // -1 casesql -2 leader 其他 work
	TaskEndTime time.Time
}

type TaskExecuteStatus struct {
	Id                int
	Version           int
	DestConnectionStr string
	DataDestination   string
	FinishStatus      bool
	DeletingStatus    bool
	LastDeletingTime  time.Time
	SchedulePeriod    int // Task
}

const (
	ip1 = "192.168.1.101"
	ip2 = "192.168.1.102"

	beforeRun = 0 //未开始
	running   = 1 //进行中
	finished  = 2 //已完成
	exception = 3 //异常错误
	timeout   = 4 // 超时

)

var WorkChannels = make(chan int)
var WorkList = make([]*WorkInfo, 0, 100)
var TaskList = make([]*TaskInfo, 0, 10)
var workIndex int

// 测试数据
var TestData TaskInfo = TaskInfo{
	Id:                 1,
	Name:               "测试33",
	SrcConnectionStr:   `{"host":"127.0.0.1","username":"root","password":"aa778899","database":"demo_mall","port":3306}`,
	DestConnectionStr:  `{"host":"120.55.97.227","port":6379,"password":"123456789"}`, // `{"url":"mongodb://admin:123456@172.16.11.84:27017","database":"data","collection":"medical"}`,
	QueryBaseSql:       "select * from admin_users",                                   //,"select distinct f_city_code,f_org_type from t_project_org_detail where f_city_code not in ('411800', '419001') order by f_city_code;",
	ScriptId:           "9eb604b62b9e02fdefc91ce7189deba9",
	LeaderIP:           "192.168.7.208",
	TotalExecutionTime: 30,
	TotalSize:          1000,
	PageSize:           200,
	StartPage:          1,
	EndPage:            5,
	DataDestination:    "redis",
	DataVersion:        0,
	ServantName:        "",
	DataProcessingType: 1,
	DataSql:            "", //"SELECT f_city_code AS cityCode, f_area_code AS areaCode, f_org_type AS orgType, f_org_name AS orgName , IFNULL(f_org_no, '') AS orgNo , IFNULL(f_chain_no, '') AS chainOrgNo , IFNULL(f_addr, '') AS orgAddr, f_lat AS lat , f_lng AS lng, f_level AS orgLevel, IFNULL(f_jump_url, '') AS jumpUrl , f_product_mode AS productMode, f_pay_mode AS payMode , CASE f_level WHEN 31 THEN 1 WHEN 32 THEN 2 WHEN 33 THEN 3 WHEN 30 THEN 4 WHEN 21 THEN 5 WHEN 22 THEN 6 WHEN 23 THEN 7 WHEN 20 THEN 8 WHEN 11 THEN 9 WHEN 12 THEN 10 WHEN 13 THEN 11 WHEN 10 THEN 12 ELSE 99 END AS orgLevelSorted, f_usable_status AS usableStatus, f_online_time AS onlineTime, f_is_bur_sement AS isEndorsed FROM t_project_org_detail WHERE (f_city_code = '${f_city_code}' or f_org_type='${f_org_type}')",
}

// 测试数据
var TestData1 TaskInfo = TaskInfo{
	Id:                 1,
	Name:               "测试33",
	SrcConnectionStr:   `{"host":"127.0.0.1","username":"root","password":"aa778899","database":"demo_mall","port":3306}`,
	DestConnectionStr:  `{"host":"127.0.0.1","port":6379,"password":"aa778899"}`,  // `{"url":"mongodb://admin:123456@172.16.11.84:27017","database":"data","collection":"medical"}`,
	QueryBaseSql:       "select username,roles from admin_users ORDER BY id desc", //,"select distinct f_city_code,f_org_type from t_project_org_detail where f_city_code not in ('411800', '419001') order by f_city_code;",
	ScriptId:           "9eb604b62b9e02fdefc91ce7189deba9",
	LeaderIP:           "192.168.7.208",
	TotalExecutionTime: 40,
	TotalSize:          100,
	PageSize:           20,
	StartPage:          1,
	EndPage:            5,
	DataDestination:    "redis",
	DataVersion:        0,
	ServantName:        "",
	DataProcessingType: 2,
	DataSql:            "select * from admin_users where (username='${username}' OR roles = '${roles}')",
	CaseSql: []string{
		"select * from admin_users where id = 100",
		"select * from admin_users where id = 101",
	},
}

func Test() string {
	log.Println("test")
	return "test"
}

func resetIndex() {
	workIndex = 0
}

func HandleReceiveTask(task TaskInfo) {
	log.Println("接收任务")

	// nextDataVersion := task.DataVersion + 1
	// 计算切片数
	pageList := getPageList(task.TotalSize, task.PageSize)
	if pageList[0] > task.StartPage || pageList[len(pageList)-1] < task.EndPage {
		log.Println("startpage or endpage is error")
		return
	}
	// 任务结束时间
	taskEndTime := time.Now().Add(time.Second * time.Duration(task.TotalExecutionTime))
	// 是否为leader节点
	isLeader := task.LeaderIP == ip1
	// 声明当前任务信息
	task.TaskEndTime = taskEndTime
	if isLeader {
		// 当前任务是否存在
		if isExistWork(task.Id, isLeader) {
			log.Printf("leader任务id= %d,ip=%s,已经存在", task.Id, task.LeaderIP)
			return
		}
		// currentTaskInfo.DataVersion = nextDataVersion
		// 存在caseSql
		if len(task.CaseSql) > 0 {
			pageList = append(pageList, -1)
		}
		WorkList = append(WorkList, &WorkInfo{
			IsLeader:    true,
			TaskInfo:    &task,
			Status:      running,
			UndonePages: pageList,
			Page:        -2, // leader任务标识
			TaskEndTime: taskEndTime,
		})
	} else {
		if isExistWork(task.Id, isLeader) {
			log.Println("这个worker任务已经存在了")
			return
		}
		// 按页码分成切片任务
		for i := task.StartPage; i <= task.EndPage; i++ {
			fmt.Println("page", i)
			WorkList = append(WorkList, &WorkInfo{
				IsLeader:    false,
				TaskInfo:    &task,
				Status:      beforeRun,
				Page:        i,
				TaskEndTime: taskEndTime,
			})
		}
		// 存在caseSql
		if len(task.CaseSql) > 0 {
			WorkList = append(WorkList, &WorkInfo{
				IsLeader:    false,
				TaskInfo:    &task,
				Status:      beforeRun,
				Page:        -1, // 存在caseSql
				TaskEndTime: taskEndTime,
			})
		}
	}
	go func() {

		// <-time.After(time.Second * time.Duration(task.TotalExecutionTime))
		<-time.After(time.Second * time.Duration(task.TotalExecutionTime))
		// if isExistWork(task.Id, isLeader) {
		log.Println("发送过期任务ID至channel!!!", task.TotalExecutionTime)

		WorkChannels <- task.Id
		// }
	}()
	var hasTask = false
	for _, v := range TaskList {
		if v.Id == task.Id {
			hasTask = true
		}
	}
	if !hasTask {
		log.Println("当前task任务已经存在")
		TaskList = append(TaskList, &task)
	}
	log.Println("worker任务分配完毕")
	log.Println("work list", WorkList)
	log.Println("task list", TaskList)

}

func isExistWork(id int, isLeader bool) bool {
	for _, v := range WorkList {
		if v.TaskInfo.Id == id && v.IsLeader == isLeader {
			return true
		}
	}
	return false
}

func getPageList(total, pageSize int) []int {
	len := total / pageSize
	if len*pageSize < total {
		len++
	}
	var list = make([]int, len, len+1)
	for i := 1; i <= len; i++ {
		list[i-1] = i
	}
	return list
}

// 主任务循环
func Run() {
	tick := time.NewTicker(5 * time.Second)
	// 启动协程 处理任务
	go func() {
		for {
			// <-tick.C
			// log.Printf("task length：%d,worker length :%d,workIndex: %d", len(TaskList), len(WorkList), workIndex)
			select {
			case <-tick.C:
				log.Printf("task length：%d,worker length :%d,workIndex: %d", len(TaskList), len(WorkList), workIndex)
			//任务超时
			case id := <-WorkChannels:
				log.Println("收到过期任务id")
				for _, v := range WorkList {
					if v.TaskInfo.Id == id {
						v.Status = timeout
					}
				}
			}
		}
	}()
	for {
		if len(WorkList) < 1 {
			continue
		}
		if workIndex >= len(WorkList) {
			workIndex = 0
		}
		currentWork := WorkList[workIndex]
		// 没有任务或者任务已经遍历结束
		if currentWork == nil {
			log.Println("没有任务或者任务已经遍历结束")
			time.Sleep(time.Second)
			resetIndex()
		} else {
			// if time.Now().After(currentWork.TaskEndTime) {
			// 	currentWork.Status = timeout
			// }
			// leader 负责监视状态
			if currentWork.IsLeader {
				switch currentWork.Status {
				case running:
					if len(currentWork.UndonePages) == 0 {
						currentWork.Status = finished
					}
				case beforeRun:
					currentWork.Status = exception
				}
				if currentWork.Status > running {
					postStatusToDriver(currentWork)
				}
			} else {
				// 执行未开始的任务
				if currentWork.Status == beforeRun {
					// 加载脚本
					// 分析工作数量
					if len(TaskList) >= 10 && len(WorkList) >= 100 {
						// 没有可用线程
						log.Println("超过最大工作数量")
					} else {
						log.Println("开启协程执行任务")
						// 开启协程执行任务
						go ExecuteWork(currentWork)
						currentWork.Status = running
					}
				}
				// 已执行不做处理
			}
			// 将完成的任务和异常超时的任务都清除掉
			if currentWork.Status > running {

				filterWorkList(workIndex)

				// leader节点直接删除taskList
				if currentWork.IsLeader {
					log.Println("leader节点直接删除taskList", currentWork)
					filterTaskkList(currentWork.TaskInfo.Id)
				} else {
					if currentWork.Status > finished {
						log.Println("worker节点只能在异常或者超时情况下删除", currentWork)
						// worker节点只能在异常或者超时情况下删除
						filterTaskkList(currentWork.TaskInfo.Id)
					}
				}
				log.Printf("任务清除 isleader:%t,id:%d,page:%d,status:%d,", currentWork.IsLeader, currentWork.TaskInfo.Id, currentWork.Page, currentWork.Status)

			} else {
				workIndex++
			}
		}

	}
}

// func reportToDriver(work *WorkInfo) {
// 	log.Printf("上报任务id:%d,状态：%d", work.TaskInfo.Id, work.Status)
// }

// delete workListItem
func filterWorkList(index int) {

	if index > -1 {
		WorkList = append(WorkList[:index], WorkList[index+1:]...)
	}
}

// delete taskListItem
func filterTaskkList(id int) {
	var index int = -1
	for i, v := range TaskList {
		if v.Id == id {
			index = i
		}
	}
	if index > -1 {
		TaskList = append(TaskList[:index], TaskList[index+1:]...)
	}
}

// 获取sql执行类型
// 定义4种查询类型
// 0 类型错误
// 1  为普通的1-1查询,需要为sql拼接limit和offset,               -   使用queryAndProcess.worker.js
// 2   为先用第一个sql的值作为key查第二个sql                     -   使用query2AndProcess.worker.js
// 3 为sql数组,便利查询sql并用脚本处理                         -   使用queryEachAndProcess.worker.js
func getSqlType(work *WorkInfo) int {
	processType := work.TaskInfo.DataProcessingType
	if work.Page == -1 {
		if len(work.TaskInfo.CaseSql) > 0 {
			return 3
		}
		return 0
	}
	if processType != 1 && processType != 2 {
		return 0
	}
	return processType

}

// 开启协程 执行任务
func ExecuteWork(work *WorkInfo) {
	defer func() {
		log.Println("ExecuteWork协程执行完毕")
	}()
	processType := getSqlType(work)
	log.Println("processType", processType)
	if processType == 0 {
		log.Println("DataProcessingType error 数据库查询方式有误")
		return
	}
	if processType == 1 {
		baseQuery(work)
	} else if processType == 2 {
		complexQuery(work)
	} else {
		eachQuery(work)
	}
}

func getPageSql(work *WorkInfo) string {
	// 计算分页
	pSize := work.TaskInfo.PageSize
	offset := (work.Page - 1) * pSize
	limit := pSize
	if offset+limit > work.TaskInfo.TotalSize {
		limit = work.TaskInfo.TotalSize - offset
	}
	// 执行sql
	sqlstr := fmt.Sprintf("%s LIMIT %d, %d", work.TaskInfo.QueryBaseSql, offset, limit)
	return sqlstr
}

// 为普通的1-1查询,需要为sql拼接limit和offset,
func baseQuery(work *WorkInfo) {
	sqlstr := getPageSql(work)
	log.Println("step1 limit sql:", sqlstr)
	db := model.GetMysqlConn(work.TaskInfo.SrcConnectionStr)
	rows, err := model.RawQueryAllField(db, sqlstr)
	if err != nil {
		work.Status = exception
		log.Println("sql 执行失败", sqlstr, err.Error())
		work.Status = exception
		return
	}
	log.Println("sql 执行完成", len(rows))
	err = postDataToLeader(work, rows)
	if err != nil {
		log.Println("postDataToLeader err", err.Error())
		work.Status = exception
		return
	}
	work.Status = finished
}

// 为先用第一个sql的值作为key查第二个sql
func complexQuery(work *WorkInfo) {
	sqlstr := getPageSql(work)
	log.Println("step2 limit sql:", sqlstr)
	db := model.GetMysqlConn(work.TaskInfo.SrcConnectionStr)
	rows, err := model.RawQueryAllField(db, sqlstr)
	if err != nil {
		log.Println("complexQuery sql1 执行失败", sqlstr, err.Error())
		work.Status = exception

		return
	}
	log.Println("complexQuery sql1 执行成功", len(rows))
	var dataSql = work.TaskInfo.DataSql
	// where语句
	whereReg := regexp.MustCompile(`where\s+\((.*?)\)`)
	whereRegMatches := whereReg.FindStringSubmatch(dataSql)
	// where语句中的动态变量
	reg := regexp.MustCompile(`\$\{(.*?)\}`)
	regMatches := reg.FindAllStringSubmatch(dataSql, -1)

	var keys = make([]string, len(regMatches))
	for i, v := range regMatches {
		keys[i] = v[1]
	}

	var str strings.Builder
	str.WriteString(" WHERE ")
	var whereSub = whereRegMatches[1]

	// 将sql1查询出来的结果构建成where后的条件语句
	// 不能用range  range是乱序的导致 i < len(rows)-1 不正确
	for i := 0; i < len(rows); i++ {

		str.WriteString(" ( ")
		var subSql = reg.ReplaceAllStringFunc(whereSub, func(s string) string {
			k := s[2 : len(s)-1]
			// fmt.Println(k)
			// return list[i]
			return rows[i][k]
		})
		str.WriteString(subSql)
		str.WriteString(" ) ")
		if i < len(rows)-1 {
			str.WriteString(" OR ")
		}
	}

	actualSql := strings.Replace(dataSql, whereRegMatches[0], str.String(), -1)
	log.Println("actualSql,", actualSql)
	rows, err = model.RawQueryAllField(db, actualSql)
	if err != nil {
		log.Println("complexQuery sql2  失败")
		work.Status = exception

		return
	}
	log.Println("complexQuery sql2 执行完成", len(rows))
	err = postDataToLeader(work, rows)
	if err != nil {
		log.Println("postDataToLeader err", err.Error())
		work.Status = exception
		return
	}
	work.Status = finished

}

// 便利查询sql查询
func eachQuery(work *WorkInfo) {
	db := model.GetMysqlConn(work.TaskInfo.SrcConnectionStr)
	var result = make(map[int]map[string]string)
	var i = 1
	for _, v := range work.TaskInfo.CaseSql {
		rows, err := model.RawQueryAllField(db, v)
		if err != nil {
			log.Println("eachQuery error", v)
			work.Status = exception
			return
		}
		for _, row := range rows {
			result[i] = row
			i++
		}
	}
	log.Println("eachQuery 执行完成", len(result))
	err := postDataToLeader(work, result)
	if err != nil {
		log.Println("postDataToLeader err", err.Error())
		work.Status = exception
		return
	}
	work.Status = finished
}

// 获取leader任务
func getLeaderWork(id int) *WorkInfo {
	for _, v := range WorkList {
		if v.IsLeader && v.TaskInfo.Id == id {
			return v
		}
	}
	return nil
}

// undonepages中移除当前任务page
func removeUndonepagesItem(page int, work *WorkInfo) {
	var index int = -1
	for i, v := range work.UndonePages {
		if v == page {
			index = i
		}
	}
	if index > -1 {
		work.UndonePages = append(work.UndonePages[:index], work.UndonePages[index+1:]...)

	}
}

// 上报数据给leader,数据为空则为上报错误状态
func postDataToLeader(work *WorkInfo, data map[int]map[string]string) error {
	// v, _ := data.(map[int]map[string]string)
	log.Printf("上报数据id=%d,page=%d,data length: %d", work.TaskInfo.Id, work.Page, len(data))
	return handlePostedData(work, data)
	// return nil
}

// 处理work节点上报的数据
func handlePostedData(work *WorkInfo, data map[int]map[string]string) error {
	leaderWorkInfo := getLeaderWork(work.TaskInfo.Id)
	if leaderWorkInfo == nil {
		log.Println("leader任务不存在！！！")
		return errors.New("leader任务不存在！！！")
	}
	defer func() {
		log.Println("处理上报数据完成！！！，undonepages", leaderWorkInfo.UndonePages)
	}()

	var postErr error
	log.Println("leaderworkinfo", leaderWorkInfo.UndonePages)
	// mongo redis的管理 避免大量的连接
	if work.TaskInfo.DataDestination == "redis" {
		// 将数据存入redis
		log.Println("将数据存入redis", work.TaskInfo.Id, work.Page)
		client := model.GetRedisConn(work.TaskInfo.DestConnectionStr, work.TaskInfo.Id, work.TaskInfo.TotalExecutionTime)
		var values = make([]string, 2*len(data))
		for i := 0; i < len(data); i++ {
			values[i*2] = fmt.Sprintf("data_%d_%d_%d_%d", work.TaskInfo.Id, work.TaskInfo.DataVersion, work.Page, i)
			v, _ := json.Marshal(data[i])
			values[i*2+1] = string(v)
		}
		// fmt.Println(values)
		start := time.Now()
		val, err := client.MSet(values).Result()
		fmt.Println("mset result", val, "error", err, time.Since(start))
		postErr = err
	} else {
		// 将数据存入mongodb
		log.Println("将数据存入mongodb")
	}
	// undonepages 移除等于work.page的元素
	removeUndonepagesItem(work.Page, leaderWorkInfo)
	return postErr

}

// 上报状态给driver节点
func postStatusToDriver(work *WorkInfo) {
	defer func() {
		log.Println("上报状态完成！！！")
	}()
	log.Printf("上报任务id:%d,状态：%d", work.TaskInfo.Id, work.Status)
	// 将任务信息存入到redis
	var data = TaskExecuteStatus{
		Id:                work.TaskInfo.Id,
		Version:           work.TaskInfo.DataVersion,
		DestConnectionStr: work.TaskInfo.DestConnectionStr,
		DataDestination:   work.TaskInfo.DataDestination,
		FinishStatus:      work.Status == finished,
		DeletingStatus:    false,
		LastDeletingTime:  time.Time{},
		SchedulePeriod:    work.TaskInfo.TotalExecutionTime, // Task
	}
	key := fmt.Sprintf(`statusInfo_%d`, work.TaskInfo.Id)
	b, err := json.Marshal(data)
	if err != nil {
		log.Println("postStatusToDriver marshal error", err)
		return
	}
	client := model.GetRedisConn(work.TaskInfo.DestConnectionStr, work.TaskInfo.Id, work.TaskInfo.TotalExecutionTime)
	// statusInfo_1 {key:value} key为版本号，超过三个就会删除最旧的版本
	res, err := client.HSet(key, fmt.Sprintf("%d", work.TaskInfo.DataVersion), string(b)).Result()
	log.Println("hset key value:", fmt.Sprintf("%d", work.TaskInfo.DataVersion), string(b))
	log.Println("set statusInfo ", res, err)
	// 执行删除操作
	ClearData(work.TaskInfo.Id)

	// 将任务完成信息存入redis 清理错误状态的任务数据清理
	// 这块感觉可以用下kv存储 问题:节点销毁后文件或许不存在了
}
